package com.jscloud.bigdata.flink.broadcast;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

/**
 * * 计数器与累加器概念
 * <p>
 * Accumulator即累加器，与Mapreduce counter的应用场景差不多，都能很好地观察task在运行期间的数据变化，可以在Flink job任务的算子函数中操作累加器，但是只能在任务执行结束之后才能获得累加器的最终结果。
 * Counter是一个具体的累加器(Accumulator)实现IntCounter, LongCounter 和 DoubleCounter。
 * ```
 * * 计数器与累加器用法
 * ~~~scala
 * (1)：创建累加器
 * IntCounter intCounter = new IntCounter();
 * (2)：注册累加器
 * getRuntimeContext.addAccumulator("num-lines",counter)
 * (3)：使用累加器
 * counter.add(1)
 * (4)：获取累加器的结果
 * myJobExecutionResult.getAccumulatorResult("num-lines")
 * ~~~
 * 分布式缓存概念
 * Flink提供了一个类似于hadoop分布式缓存，可以使用户在并行函数中很方便的读取本地文件。
 * 前面讲到的广播变量是将一些共享的数据放在TaskManager内存中，而Distribute cache是从外部加载一个文件/目录(例如hdfs)，然后分别复制到每一个TaskManager的本地磁盘中。
 * 分布式缓存用法：
 * (1)：使用Flink运行环境调用registerCachedFile注册一个分布式缓存
 * env.registerCachedFile("hdfs:///path/to/your/file", "hdfsFile")
 * (2): 获取分布式缓存
 * File myFile = getRuntimeContext().getDistributedCache().getFile("hdfsFile");
 */
public class FlinkBroadCastByJava {
        public static void main(String[] args) throws Exception {
                ExecutionEnvironment executionEnvironment = ExecutionEnvironment.getExecutionEnvironment();

                //用户姓名以及年龄
                ArrayList<String> arrayList = new ArrayList<>();
                arrayList.add("zhangsan 30");
                arrayList.add("lisi 28");
                arrayList.add("wangwu 18");

                //将用户住址信息给广播出去
                ArrayList<String> arrayList2 = new ArrayList<>();
                arrayList2.add("zhangsan beijing");
                arrayList2.add("lisi shanghai");
                arrayList2.add("wangwu shenzhen");

                DataSource<ArrayList<String>> userAndAge = executionEnvironment.fromElements(arrayList);

                DataSource<String> stringDataSource = executionEnvironment.fromCollection(arrayList2);

                FlatMapOperator<ArrayList<String>, String> result = userAndAge.flatMap(new RichFlatMapFunction<ArrayList<String>, String>() {
                        private HashMap<String, String> userAddress = new HashMap<String, String>();

                        @Override
                        public void open(Configuration parameters) throws Exception {
                                List<String> userAndAddress = getRuntimeContext().getBroadcastVariable("userAndAddress");
                                for (String andAddress : userAndAddress) {
                                        String[] s = andAddress.split(" ");
                                        userAddress.put(s[0], s[1]);
                                }
                        }

                        @Override
                        public void flatMap(ArrayList<String> value, Collector<String> out) throws Exception {
                                for (String s : value) {
                                        String[] s1 = s.split(" ");
                                        String age = userAddress.get(s1[0]);
                                        out.collect(s1[0] + "\t" + s1[1] + "\t" + age);
                                }
                        }

                }).withBroadcastSet(stringDataSource, "userAndAddress");
                result.print();
        }
}